package day19;

import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.core.fs.FileSystem;

import java.io.File;

public class InputOutputTest {
    public static void main(String[] args) throws Exception {
        /*
        * spark中生产rdd数据有三种：
        * 1.从外部存储介质中读取textFile。
        * 2.并行化方式生成RDD。
        * 3.从hive表中获取，用spark来读取hive表。(用hiveOnSpark的方式,读取用hiveOnContext实例来读取？)
        * */


        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        // 1.flink输入数据的方式：1.自己写数据，2.读取本地文件数据 ，3.读取hdfs数据 ，4.generateSequence(a,b)创建数据
        DataSet<String> text = env.fromElements("hello fuhao","hello java");
        DataSet<String> localText = env.readTextFile("C:\\Users\\fuhao\\Desktop\\one.txt");
        DataSet<String> hdfsText = env.readTextFile("hdfs://master:9000/Wordcount3/input");
        DataSet<Long> numbers = env.generateSequence(1,10);
        text.print();
        localText.print();
        hdfsText.print();
        numbers.print();

        // 2.flink输出数据的方式：1.writeAsText("E:\\data")写到本地，2.写到hdfs系统
//        hdfsText.writeAsText("E:\\data3");  //文件有三个，然后内部分了4个区，所以最终保存成四个文件
        //hdfsText.writeAsText("E:\\data2", FileSystem.WriteMode.OVERWRITE);//覆盖的模式写。

        //写到hdfs上必须要先创建文件目录。并且必须有些的权限：hadoop fs -chmod -R 777 /data2
        hdfsText.writeAsText("hdfs://master:9000/data4", FileSystem.WriteMode.OVERWRITE);

        env.execute();//写入文件好像必须要这行代码

    }
}
